package com.qing.write

import java.util
import java.util.UUID
import java.util.regex.Pattern

import com.qing.HTableUtil
import org.apache.hadoop.conf.Configuration
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.hadoop.hbase.client.HTable
import org.apache.hadoop.hbase.client.Put
import org.apache.hadoop.hbase.util.Bytes

/**
  * Created by wuliao on 2017/9/12.
  */
object Parse2Hbase {
  def main(args: Array[String]): Unit = {
    val path: String = "/Users/wuliao/Desktop/messagelog/*/"
    val conf = new SparkConf()
      .setAppName("hello")
      .set("spark.driver.cores", "1")
      .set("spark.driver.memory", "4096m")
      .set("spark.executor.memory", "4096m")
      .setMaster("local")
    val sc = new SparkContext(conf)
    val rdd = sc.textFile(path, 10)

    rdd.mapPartitions(it => {
      val puts = new util.ArrayList[Put]()
      while (it.hasNext) {
        val s = it.next
        parse(s, puts)
      }
      val table = HTableUtil.getHTable("testtable")
      table.put(puts)
      table.close()
      it
    }).count()
  }

  def parse(s: String, puts: util.ArrayList[Put]): Unit = {
    val pattern = "messages.*:(\\w{3} {1,2}\\d{1,2} \\d{2}:\\d{2}:\\d{2}) {1,2}([\\s\\S]+?) {1,2}([\\s\\S]+)\\[([0-9]+)\\]: {1,2}([\\s\\S]+)";
    val r = Pattern.compile(pattern)
    val m = r.matcher(s)
    if (m.find) {
      val itemId = UUID.randomUUID().toString()
      val put = new Put(Bytes.toBytes(itemId))
      val date = m.group(1)
      val month = getMonth(date.substring(0, 3)) + 1
      val day = date.substring(4, 6)
      val time = date.substring(7).replaceAll(":", "")
      var monthStr = ""
      if (month < 9) {
        monthStr = "0" + month
      } else {
        monthStr = month + ""
      }
      val c = "2017" + monthStr + day + time
      put.addColumn("log".getBytes, "time".getBytes, c.getBytes)

      val host = m.group(2)
      put.addColumn("log".getBytes, "host".getBytes, host.getBytes)

      val user = m.group(3)
      put.addColumn("log".getBytes, "user".getBytes, user.getBytes)

      val id = m.group(4)
      put.addColumn("log".getBytes, "id".getBytes, id.getBytes)

      val reason = m.group(5)
      put.addColumn("log".getBytes, "reason".getBytes, reason.getBytes)

      val p = Pattern.compile("[\\s\\S]+?(\\d+.\\d+.\\d+.\\d+:\\d+)[\\s\\S]+")
      val matcher = p.matcher(reason)
      if (matcher.find) {
        val ipAndPort = matcher.group(1)
        val ip = ipAndPort.split(":")(0)
        put.addColumn("log".getBytes, "ip".getBytes, ip.getBytes)
        val port = ipAndPort.split(":")(1)
        put.addColumn("log".getBytes, "port".getBytes, port.getBytes)
      }
      puts.add(put)
    } else {
      println(s)
    }
  }

  def getMonth(s: String): Int = {
    val months = Array[String]("Jan", "Feb", "Mar", "Apr", "May", "Jun", "Jul", "Aug", "Sep", "Oct", "Nov", "Dec")
    val size = months.length
    for (i <- 0 until size) {
      if (months.apply(i).equals(s)) {
        return i
      }
    }
    0
  }


}
